package com.project.pulsar.dealTopic;

import com.project.pulsar.conf.PulsarConf;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

/**
 * 死信队列
 *
 */
@Slf4j
@RestController
public class DealTopicController {
    @Autowired
    PulsarConf pulsarConf;


    /**
     * 同步-生产消息
     * @param msg
     * @throws PulsarClientException
     */
    @GetMapping("/deal/sendMsgBySync")
    public MessageId sendMsgBySync(String msg) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();

        Producer<byte[]> producer = pulsarFactory.newProducer()
                .topic("my-deal-topic")
                .create();
        // 然后你就可以发送消息到指定的broker 和topic上：
        return producer.send(msg.getBytes());
    }

    @Bean
    public void comsumerByDealListener() throws Exception {
        MessageListener myMessageListener = (consumer, msg) -> {
            String anObject = new String(msg.getData());
            System.out.println("收到消息了");
           // int i=1/0;//故意报错
            try {
                System.out.println("Message received: " + anObject);
//                consumer.acknowledge(msg);//故意不确认消费
            } catch (Exception e) {
//                consumer.negativeAcknowledge(msg);//故意不确认消费
            }
        };
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        pulsarFactory.newConsumer()
                .topic("my-deal-topic")
                .subscriptionName("my-subscriptionByListener")
                .ackTimeout(3, TimeUnit.SECONDS)//指定超时多久未确认消息后判断为异常
                .subscriptionType(SubscriptionType.Shared)
                .messageListener(myMessageListener)
                .deadLetterPolicy(DeadLetterPolicy.builder()
                        .maxRedeliverCount(4)//指定消费失败重试次数，超过次数进入死信队列
                        .deadLetterTopic("dealtopic")//死信队列名，不指定的话默认：{TopicName}-{Subscription}-DLQ
                        .build())
                .subscribe();
    }
    @GetMapping("/deal/cunsumerDeadLetterTopic")
    public void cunsumerDeadLetterTopic() throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic("dealtopic")
                .subscriptionName("mysubscriptiondlq")
                .subscribe();
        Message<byte[]> receive = consumer.receive();
        System.out.println(new String(receive.getData()));
        consumer.acknowledge(receive);//确认消息被消费
        consumer.close();
    }




}
